package cn.lsh.main;

import cn.lsh.comparator.Friend2GroupComparator;
import cn.lsh.comparator.Friend2SortComparator;
import cn.lsh.mapper.Friend2Mapper;
import cn.lsh.reducer.Friend2Reducer;
import cn.lsh.writable.FriendRelation;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;

/**
 * 在friend输出好友间接关系的前提下输出各用户的最佳推荐好友
 * 数据：
 * cat:hadoop	2
 * cat:hello	2
 * cat:mr	1
 * cat:word	1
 * hadoop:hello	3
 * hadoop:mr	1
 * hive:tom	3
 * mr:tom	1
 * mr:word	2
 * tom:word	2
 *
 * 分析：
 * 1、map对记录映射成正反两组
 * 2、排序比价器根据第一个用户和关系数倒排序
 * 3、分组比较器根据第一个用户分组
 * 4、reduce取出关系数最大的推荐关系
 */
public class Friend2Driver extends BaseDriver {

	public static void main(String[] args) throws Exception {
		buildJob(true);

		// job.setInputFormatClass(Friend2Mapper.class);
		job.setMapperClass(Friend2Mapper.class);
		job.setMapOutputKeyClass(FriendRelation.class);
		job.setMapOutputValueClass(IntWritable.class);

		job.setSortComparatorClass(Friend2SortComparator.class);

		job.setGroupingComparatorClass(Friend2GroupComparator.class);

		job.setReducerClass(Friend2Reducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);

		args = new String[] {"/test/output/friend/part-r-00000", "/test/output/firend2"};
		setInputAndOutPut(args, true);

		job.waitForCompletion(true);
	}
}
